package org.xcorpio.java8.chapter07;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream;

public class ForkJoinSumCalculator extends RecursiveTask<Long> {

    private static final long serialVersionUID = 7588117855462582991L;
    private final long[] numbers;
    private final int start;
    private final int end;

    public static final long THRESHOLD = 4;

    /**
     * Public constructor used to create the main task.
     * 
     * @param numbers
     */
    public ForkJoinSumCalculator(long[] numbers) {
        this(numbers, 0, numbers.length);
    }

    /**
     * Private constructor used to recursively create subtasks of the main task.
     * 
     * @param numbers
     * @param start
     * @param end
     */
    private ForkJoinSumCalculator(long[] numbers, int start, int end) {
        this.numbers = numbers;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        /**
         * The size of the portion of the array summed by this task.
         */
        int length = this.end - this.start;
        /**
         * If the size is less than or equal to the threshold, compute the
         * result sequentially.
         */
        if (length <= THRESHOLD) {
            return computeSequentially();
        } else {
            /**
             * Create a subtask to sum the first half of the array.
             */
            ForkJoinSumCalculator leftTask = new ForkJoinSumCalculator(this.numbers, this.start,
                    this.start + length / 2);
            /**
             * Asynchronously execute the newly created subtask using another
             * thread of the ForkJoinPool
             */
            leftTask.fork();
            /**
             * Create a subtask to sum the second half of the array.
             */
            ForkJoinSumCalculator rightTask = new ForkJoinSumCalculator(this.numbers, this.start + length / 2,
                    this.end);
            /**
             * Execute this second subtask synchronously, potentially allowing
             * further recursive splits.
             */
            Long rightResult = rightTask.compute();
            /**
             * Read the result of the first subtask or wait for it if it isn't
             * ready.
             */
            Long leftResult = leftTask.join();
            return leftResult + rightResult;
        }
    }

    /**
     * Simple algorithm calculating the result of a subtask when it's no longer
     * divisible.
     * 
     * @return
     */
    private Long computeSequentially() {
        long sum = 0;
        for (int i = this.start; i < this.end; ++i) {
            sum += this.numbers[i];
        }
        return sum;
    }

    public static void main(String[] args) {
        int num = 32;
        long[] numbers = LongStream.rangeClosed(1, num).toArray();
        ForkJoinSumCalculator task = new ForkJoinSumCalculator(numbers);
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        Long result = forkJoinPool.invoke(task);
        System.out.println(result);
    }
}
